package com.george.rocketmq.service;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

/**
 * @title: RocketMqConsumerService.java
 * @description: RocketMq消息消费者
 * @author: George
 * @date: 2023/3/24 10:59
 */
@Slf4j
@Component
public class RocketMqConsumerService {

    /**
     * 消费者这里要注意设置的参数要正确，我这里为了方便就写在一个类里
     * 这里消费者会直接监听生产者发送的消息，一旦生产者那边发送消息，对应这里就会消费
     * 另外可以对注解 @RocketMQMessageListener 点进去看看它的属性参数，都是非常熟悉的
     * 还有，实际生产中，应避免自己的消费者代码出现非业务逻辑上的错误，比如消费时某个消费者报类型转换异常，建议多用TAG做区分
     */


    /**
     * topic需要和生产者的topic一致，consumerGroup属性是必须指定的，内容可以随意
     * selectorExpression的意思指的就是tag，默认为“*”，不设置的话会监听所有消息
     *
     * @param null
     * @author George
     * @date 2023/3/24 11:04
     * @return
     */
    @Service
    @RocketMQMessageListener(topic = "RLT_TEST_TOPIC", selectorExpression = "tag1", consumerGroup = "Con_Group_One")
    public class ConsumerSend implements RocketMQListener<Object> {
        // 监听到消息就会执行此方法
        @Override
        public void onMessage(Object object) {
            log.info("监听到消息：object={}", JSON.toJSONString(object));
        }
    }

    /**
     * 注意：这个ConsumerSend2和上面ConsumerSend在没有添加tag做区分时，不能共存，
     * 不然生产者发送一条消息，这两个都会去消费，如果类型不同会有一个报错，所以实际运用中最好加上tag，写这只是让你看知道就行
     *
     * @param null
     * @author George
     * @date 2023/3/24 11:04
     * @return
     */
    @Service
    @RocketMQMessageListener(topic = "RLT_TEST_TOPIC", consumerGroup = "Con_Group_Two")
    public class ConsumerSend2 implements RocketMQListener<String> {
        @Override
        public void onMessage(String str) {
            log.info("监听到消息：str={}", str);
        }
    }

    /**
     * MessageExt：是一个消息接收通配符，不管发送的是String还是对象，都可接收，当然也可以像上面明确指定类型（我建议还是指定类型较方便）
     *
     * @param null
     * @author George
     * @date 2023/3/24 11:05
     * @return
     */
    @Service
    @RocketMQMessageListener(topic = "RLT_TEST_TOPIC", selectorExpression = "tag2", consumerGroup = "Con_Group_Three")
    public class Consumer implements RocketMQListener<MessageExt> {
        @Override
        public void onMessage(MessageExt messageExt) {
            byte[] body = messageExt.getBody();
            String msg = new String(body);
            log.info("监听到消息：msg={}", msg);
        }
    }

}
